package com.seezoon.infrastructure.tcp.handler;

import com.seezoon.infrastructure.tcp.codec.JT808;
import com.seezoon.infrastructure.tcp.codec.JTMessage;
import com.seezoon.infrastructure.tcp.codec.JTSerialization;
import com.seezoon.infrastructure.tcp.jt808.JT0005;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Sharable
public class SubpacketHandler extends SimpleChannelInboundHandler<JTMessage> {

    private static final long SESSION_TIMEOUT = 5 * 60 * 1000; // 5分钟超时
    // 会话重组器缓存 <sessionKey, reassembler>
    private final Map<String, SubpacketSession> sessionMap = new ConcurrentHashMap<>();
    private final ScheduledExecutorService scheduler;
    private final String threadName = "TcpSubpackSession-task";
    private final long interval = Duration.ofMinutes(1).toMillis();

    public SubpacketHandler() {
        scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, threadName);
            thread.setDaemon(true);
            return thread;
        });

        scheduler.scheduleWithFixedDelay(() -> {
            try {
                // 清理
                cleanupExpiredSessions();
                // 重传
                requestMissingPackets();
            } catch (Throwable e) {
                log.error(threadName + " sdk scheduler error", e);
            }
        }, interval, interval, TimeUnit.MILLISECONDS);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, JTMessage msg) throws Exception {
        // 非分包消息直接传递
        if (!msg.hasSubPackage()) {
            ctx.fireChannelRead(msg);
            return;
        }

        // 生成会话Key
        String sessionKey = generateSessionKey(ctx.channel(), msg);

        // 获取或创建会话重组器
        SubpacketSession session = sessionMap.computeIfAbsent(sessionKey,
                k -> new SubpacketSession(msg.getPackageTotal(), msg.getSerialNo(), msg.getMsgSerialNo(),
                        ctx.channel()));

        // 添加分包
        session.addPacket(msg.getPackageNo(), msg);

        // 检查是否完成
        if (session.isComplete()) {
            // 组装完整消息
            JTMessage fullMessage = assembleFullMessage(session);
            sessionMap.remove(sessionKey);
            ctx.fireChannelRead(fullMessage);
        }
    }

    private String generateSessionKey(Channel channel, JTMessage msg) {
        return channel.id().asLongText() + "|" + msg.getSerialNo() + "|" + msg.getMsgId() + "|" + msg.getMsgSerialNo();
    }

    private JTMessage assembleFullMessage(SubpacketSession session) {
        JTMessage firstPacket = session.getPacket(1);
        JTMessage fullMessage = new JTMessage();

        // 复制公共属性
        fullMessage.setMsgId(firstPacket.getMsgId());
        fullMessage.setMsgProperties(firstPacket.getMsgProperties() & ~(1 << 13)); // 清除分包标志
        fullMessage.setSerialNo(firstPacket.getSerialNo());
        fullMessage.setMsgSerialNo(firstPacket.getMsgSerialNo());

        // 组装消息体
        ByteBuf fullBody = Unpooled.compositeBuffer();
        for (int i = 1; i <= session.totalPackets; i++) {
            JTMessage packet = session.getPacket(i);
            fullBody.writeBytes(packet.getBody());
        }
        fullMessage.setBody(fullBody);

        return fullMessage;
    }

    // 定时清理过期会话
    public void cleanupExpiredSessions() {
        long now = System.currentTimeMillis();
        sessionMap.entrySet().removeIf(entry ->
                (now - entry.getValue().lastAccessTime) > SESSION_TIMEOUT
        );
    }

    // 重传
    public void requestMissingPackets() {
        sessionMap.entrySet().forEach(entry -> {
            SubpacketSession session = entry.getValue();
            // 检查超时
            if (System.currentTimeMillis() - session.lastAccessTime < SESSION_TIMEOUT / 2) {
                return;
            }
            String serialNo = session.serialNo;
            int msgSerialNo = session.msgSerialNo;

            // 重传
            for (int i = 1; i <= session.totalPackets; i++) {
                if (session.getPacket(i) != null) {
                    continue;
                }
                JTMessage jtMessage = new JTMessage();
                JT0005 request = new JT0005();
                //原始消息流水号
                request.setOriginalMsgSerialNo(session.msgSerialNo);
                //重传包总数
                request.setRetransmitPackageTotal(1);
                //重传包ID列表
                request.setRetransmitPackageNos(new int[]{i});
                // 重传请求消息ID
                jtMessage.setMsgId(JT808.服务器补传分包请求);
                jtMessage.setSerialNo(serialNo);
                jtMessage.setMsgSerialNo(session.nextSequence());
                byte[] encoded = JTSerialization.encode(request);
                jtMessage.setMsgProperties(encoded.length);
                jtMessage.setBody(Unpooled.wrappedBuffer(encoded));

                if (session.channel.isActive()) {
                    session.channel.writeAndFlush(jtMessage);
                    log.info("Sent retransmit jtMessage for serialNo:{},msgSerialNo:{}, packet {}/{}", serialNo,
                            msgSerialNo, i, session.totalPackets);
                    return;
                } else {
                    log.info("Sent retransmit jtMessage failed for serialNo:{},msgSerialNo:{}, packet {}/{}", serialNo,
                            msgSerialNo, i, session.totalPackets);
                }
            }
        });
    }

    private static class SubpacketSession {

        private final int totalPackets;
        /**
         * 终端编号
         */
        private final String serialNo;
        /**
         * 消息流水号
         */
        private final int msgSerialNo;
        private final JTMessage[] packets;
        private final Channel channel;
        private final AtomicInteger sequence = new AtomicInteger(0);
        private long lastAccessTime = System.currentTimeMillis();

        public SubpacketSession(int totalPackets, String serialNo, int msgSerialNo, Channel channel) {
            this.totalPackets = totalPackets;
            this.serialNo = serialNo;
            this.msgSerialNo = msgSerialNo;
            this.packets = new JTMessage[totalPackets + 1]; // 索引从1开始
            this.channel = channel;
        }

        public void addPacket(int packetNo, JTMessage packet) {
            if (packetNo < 1 || packetNo > totalPackets) {
                throw new IllegalArgumentException(
                        "Invalid packet number: " + packetNo + " totalPackets:" + totalPackets);
            }
            packets[packetNo] = packet;
            lastAccessTime = System.currentTimeMillis();
        }

        public JTMessage getPacket(int packetNo) {
            return packets[packetNo];
        }

        public boolean isComplete() {
            for (int i = 1; i <= totalPackets; i++) {
                if (packets[i] == null) {
                    return false;
                }
            }
            return true;
        }

        public int nextSequence() {
            // 无符号short 最大值后，从0 开始
            return sequence.updateAndGet(current ->
                    (current >= 65535) ? 0 : current + 1
            );
        }
    }
}
